[#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker #28548

baeminbo · 2023-09-20T02:12:09Z

This is another case of #20970 (harmless error logs with gRPC leak).

BigQueryIO.read() with Method.DIRECT_READ in Dataflow streaming jobs can cause Dataflow to output the gRPC leak error logs like below.

This is because getCheckpointMark() in BoundedToUnboundedSourceAdapter calls init() where residualSource field is updated, but it missed closing old residualSource, which caused a leak of ResidualSource#reader closing. The reader in the stacktrace below is BigQueryStorageStreamReader. Not closing BigQueryStorageStreamReader causes the gRPC leak error logs.

Error mesasge:

*~*~*~ Previous channel ManagedChannelImpl{logId=9, target=bigquerystorage.googleapis.com:443} was not shutdown properly!!! ~*~*~*
    Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true.

Stacktrace:

java.lang.RuntimeException: ManagedChannel allocation site
	at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:102)
	at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:60)
	at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:51)
	at io.grpc.internal.ManagedChannelImplBuilder.build(ManagedChannelImplBuilder.java:631)
	at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:297)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:391)
	at com.google.api.gax.grpc.ChannelPool.<init>(ChannelPool.java:107)
	at com.google.api.gax.grpc.ChannelPool.create(ChannelPool.java:85)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:237)
	at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:231)
	at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:236)
	at com.google.cloud.bigquery.storage.v1.stub.EnhancedBigQueryReadStub.create(EnhancedBigQueryReadStub.java:99)
	at com.google.cloud.bigquery.storage.v1.BigQueryReadClient.<init>(BigQueryReadClient.java:130)
	at com.google.cloud.bigquery.storage.v1.BigQueryReadClient.create(BigQueryReadClient.java:110)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$StorageClientImpl.<init>(BigQueryServicesImpl.java:1667)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl$StorageClientImpl.<init>(BigQueryServicesImpl.java:1598)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryServicesImpl.getStorageClient(BigQueryServicesImpl.java:201)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryStorageStreamSource$BigQueryStorageStreamReader.<init>(BigQueryStorageStreamSource.java:194)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryStorageStreamSource$BigQueryStorageStreamReader.<init>(BigQueryStorageStreamSource.java:147)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryStorageStreamSource.createReader(BigQueryStorageStreamSource.java:138)
	at org.apache.beam.sdk.io.gcp.bigquery.BigQueryStorageStreamSource.createReader(BigQueryStorageStreamSource.java:56)
	at org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter$ResidualSource.advance(UnboundedReadFromBoundedSource.java:473)
	at org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter$ResidualSource.access$300(UnboundedReadFromBoundedSource.java:452)
	at org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter$Reader.advance(UnboundedReadFromBoundedSource.java:304)
	at org.apache.beam.runners.core.construction.UnboundedReadFromBoundedSource$BoundedToUnboundedSourceAdapter$Reader.start(UnboundedReadFromBoundedSource.java:297)
	at org.apache.beam.runners.dataflow.worker.WorkerCustomSources$UnboundedReaderIterator.start(WorkerCustomSources.java:816)
	at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation$SynchronizedReaderIterator.start(ReadOperation.java:381)
	at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:211)
	at org.apache.beam.runners.dataflow.worker.util.common.worker.ReadOperation.start(ReadOperation.java:169)
	at org.apache.beam.runners.dataflow.worker.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:83)
	at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.process(StreamingDataflowWorker.java:1433)
	at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker.access$800(StreamingDataflowWorker.java:155)
	at org.apache.beam.runners.dataflow.worker.StreamingDataflowWorker$4.run(StreamingDataflowWorker.java:1056)
	at org.apache.beam.runners.dataflow.worker.util.BoundedQueueExecutor.lambda$executeLockHeld$0(BoundedQueueExecutor.java:163)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:834)

github-actions · 2023-09-20T03:34:46Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

baeminbo · 2023-09-22T07:03:19Z

I put a comment not owned to the reader field in the class UnboundedReaderIterator in WorkerCustomSources.java. My first fix was simply to close the reader in UnboundedReaderIterator#close(). But, it had a bug to read the same data again from a reader in integration tests. To avoid this mistake again, I put the comment and additional check at ResidualSource#getCheckpointMark()

baeminbo · 2023-09-24T02:19:52Z

retest this please.

liferoad · 2023-09-24T14:26:31Z

Run Java_GCP_IO_Direct PreCommit

liferoad · 2023-09-24T14:27:07Z

R: @kennknowles

liferoad · 2023-09-24T14:27:27Z

R: @Abacn

github-actions · 2023-09-24T14:29:14Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

kennknowles · 2023-09-26T12:55:55Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

@@ -288,6 +288,15 @@ private void init(
            residualElementsList == null
                ? new ResidualElements(Collections.emptyList())
                : new ResidualElements(residualElementsList);
+
+        if (this.residualSource != null) {


Just checking - this means that we are calling init() again without ever calling close() on the reader?

Yes, it's at Reader.getCheckpointMark().

The getCheckpointMark() is called at bundle finish in Dataflow streaming jobs, and close() will be pending as readers are cached. A reader can be reused in next bundles.

Oh yea based on that the change is obviously correct. Thanks!

kennknowles · 2023-09-26T12:57:02Z

I can see that testing a fix for a memory leak may be challenging, but can we do anything to test this change?

baeminbo

I'll try to make a test to confirm the effect of this change :) I'll ping when it's ready.

baeminbo · 2023-09-27T02:33:26Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

@@ -288,6 +288,15 @@ private void init(
            residualElementsList == null
                ? new ResidualElements(Collections.emptyList())
                : new ResidualElements(residualElementsList);
+
+        if (this.residualSource != null) {


Yes, it's at Reader.getCheckpointMark().

The getCheckpointMark() is called at bundle finish in Dataflow streaming jobs, and close() will be pending as readers are cached. A reader can be reused in next bundles.

kennknowles · 2023-09-27T13:52:11Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

@@ -288,6 +288,15 @@ private void init(
            residualElementsList == null
                ? new ResidualElements(Collections.emptyList())
                : new ResidualElements(residualElementsList);
+
+        if (this.residualSource != null) {


Oh yea based on that the change is obviously correct. Thanks!

kennknowles · 2023-09-27T13:52:46Z

.../src/main/java/org/apache/beam/runners/core/construction/UnboundedReadFromBoundedSource.java

@@ -505,6 +514,7 @@ BoundedSource<T> getSource() {
      }

      Checkpoint<T> getCheckpointMark() {
+        checkArgument(!closed, "getCheckpointMark() call on closed %s", getClass().getName());


This should be checkState since there are no arguments.

kennknowles · 2023-09-27T13:54:34Z

...w-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WorkerCustomSources.java

@@ -776,7 +776,7 @@ public double getRemainingParallelism() {

  private static class UnboundedReaderIterator<T>
      extends NativeReader.NativeReaderIterator<WindowedValue<ValueWithRecordId<T>>> {
-    private final UnboundedSource.UnboundedReader<T> reader;
+    private final UnboundedSource.UnboundedReader<T> reader; // not owned


What does ownership mean here? Not to do with memory management obviously. If I understand correctly, it means that there may be other calls to various state methods so you can never assume what state the reader will be in? (it would be worth writing the whole long comment about what this means here)

Thanks, I added a comment to explain it in detail, rather than a short ambiguous comment.

baeminbo · 2023-09-29T12:40:48Z

@kennknowles I added a unit test UnboundedReadFromBoundedSourceTest#testReadersClosedProperly to check reader closing leak. Could you review the change? Thank you.

kennknowles · 2023-09-29T15:23:06Z

I feel like I want to run a bit of a variety of integration tests but I'm not 100% sure which ones.

kennknowles · 2023-09-29T16:44:54Z

Run Java_IOs_Direct PreCommit

kennknowles · 2023-10-05T18:47:53Z

OK the failure we see has been fixed at HEAD. Can you rebase against the master branch and we can re-run to get green.

baeminbo · 2023-10-06T03:12:55Z

Run Java_GCP_IO_Direct PreCommit

baeminbo · 2023-10-11T01:59:07Z

Run Java PreCommit

…er#init()

…ReaderIterator.reader

…rceAdapter.Reader

baeminbo · 2023-10-11T04:26:09Z

Run Java_GCP_IO_Direct PreCommit

baeminbo · 2023-10-11T04:56:04Z

Run Java PreCommit

baeminbo · 2023-10-11T06:07:03Z

Run Java_GCP_IO_Direct PreCommit

baeminbo · 2023-10-11T07:11:45Z

@kennknowles Some tests were flaky. But, finally, it passed all the checks.

github-actions bot added runners dataflow labels Sep 20, 2023

baeminbo force-pushed the grpc-leak branch 2 times, most recently from 01b3774 to 204c697 Compare September 22, 2023 06:50

github-actions bot added the core label Sep 22, 2023

baeminbo changed the title ~~[#20970] Fix gRPC leak by closing reader at WorkerCustomSources.UnboundedReaderIterator in Dataflow worker~~ [#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker Sep 22, 2023

kennknowles reviewed Sep 26, 2023

View reviewed changes

baeminbo commented Sep 27, 2023

View reviewed changes

kennknowles reviewed Sep 27, 2023

View reviewed changes

kennknowles approved these changes Sep 29, 2023

View reviewed changes

baeminbo force-pushed the grpc-leak branch from ea5a520 to 207bee8 Compare October 6, 2023 01:45

baeminbo added 3 commits October 10, 2023 20:06

Close existing ResidualSource at BoundedToUnboundedSourceAdapter.Read…

1f1ad9a

…er#init()

Change Precondition check method and clarify comments about Unbounded…

061239c

…ReaderIterator.reader

Add a unit test to check reader closing leak in BoundedToUnboundedSou…

a20141f

…rceAdapter.Reader

baeminbo force-pushed the grpc-leak branch from 207bee8 to a20141f Compare October 11, 2023 03:06

kennknowles merged commit aedfa46 into apache:master Oct 11, 2023
21 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker #28548

[#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker #28548

baeminbo commented Sep 20, 2023 •

edited

Loading

github-actions bot commented Sep 20, 2023

baeminbo commented Sep 22, 2023

baeminbo commented Sep 24, 2023

liferoad commented Sep 24, 2023

liferoad commented Sep 24, 2023

liferoad commented Sep 24, 2023

github-actions bot commented Sep 24, 2023

kennknowles Sep 26, 2023

baeminbo Sep 27, 2023

kennknowles Sep 27, 2023

kennknowles commented Sep 26, 2023

baeminbo left a comment

baeminbo Sep 27, 2023

kennknowles Sep 27, 2023

kennknowles Sep 27, 2023

baeminbo Sep 29, 2023

kennknowles Sep 27, 2023

baeminbo Sep 29, 2023

baeminbo commented Sep 29, 2023

kennknowles commented Sep 29, 2023

kennknowles commented Sep 29, 2023

kennknowles commented Oct 5, 2023

baeminbo commented Oct 6, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

[#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker #28548

[#20970] Fix gRPC leak by closing ResidualSource at BoundedToUnboundedSourceAdapter.Reader#init() in Dataflow worker #28548

Conversation

baeminbo commented Sep 20, 2023 • edited Loading

github-actions bot commented Sep 20, 2023

baeminbo commented Sep 22, 2023

baeminbo commented Sep 24, 2023

liferoad commented Sep 24, 2023

liferoad commented Sep 24, 2023

liferoad commented Sep 24, 2023

github-actions bot commented Sep 24, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Sep 26, 2023

baeminbo left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

baeminbo commented Sep 29, 2023

kennknowles commented Sep 29, 2023

kennknowles commented Sep 29, 2023

kennknowles commented Oct 5, 2023

baeminbo commented Oct 6, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Oct 11, 2023

baeminbo commented Sep 20, 2023 •

edited

Loading